package com.lsx143.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.app.BaseAppV2;
import com.lsx143.realtime.bean.OrderWide;
import com.lsx143.realtime.bean.PaymentWide;
import com.lsx143.realtime.bean.ProductStats;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.function.DimAsyncFunction;
import com.lsx143.realtime.util.DimUtil;
import com.lsx143.realtime.util.MyUtil;
import com.lsx143.realtime.util.SinkUtil;
import com.lsx143.realtime.util.WriteUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class DWSProductStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DWSProductStatsApp().init(
                30002,
                "DWSProductStatsApp",
                1,
                "DWSProductStatsApp",
                Constants.TOPIC_DWD_PAGE_LOG,
                Constants.TOPIC_DWD_DISPLAY_LOG,
                Constants.TOPIC_DWD_FAVOR_INFO,
                Constants.TOPIC_DWD_CART_INFO,
                Constants.TOPIC_DWM_ORDER_WIDE,
                Constants.TOPIC_DWM_PAYMENT_WIDE,
                Constants.TOPIC_DWD_REFUND_PAYMENT,
                Constants.TOPIC_DWD_COMMENT_INFO
        );
    }

    @Override
    protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> srcStreams) {
        //1.解析转换,union流
        DataStream<ProductStats> unionStream = parseStreamAndUnion(srcStreams);

        //2.开窗聚合
        SingleOutputStreamOperator<ProductStats> reducedStream = aggregateStream(unionStream);

        //3.补充维度信息
        SingleOutputStreamOperator<ProductStats> withDimStream = fillUpDim(reducedStream);
//        withDimStream.print();

        //4.写出到clickhouse
        WriteUtil.writeToClickHouse(withDimStream, Constants.CLICKHOUSE_TABLE_PRODUCT_STATS, ProductStats.class);
    }


    private SingleOutputStreamOperator<ProductStats> fillUpDim(SingleOutputStreamOperator<ProductStats> reducedStream) {
        return AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    protected void addDim(Connection phoenixCon,
                                          Jedis jedis,
                                          ProductStats ps,
                                          ResultFuture<ProductStats> resultFuture) {
                        //1.sku
                        JSONObject skuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SKU_INFO, ps.getSku_id());
                        ps.setSku_name(skuInfo.getString("SKU_NAME"));
                        ps.setSku_price(skuInfo.getBigDecimal("PRICE"));
                        ps.setSpu_id(skuInfo.getLong("SPU_ID"));
                        ps.setTm_id(skuInfo.getLong("TM_ID"));
                        ps.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                        //2.spu
                        JSONObject spuInfo = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_SPU_INFO, ps.getSpu_id());
                        ps.setSpu_name(spuInfo.getString("SPU_NAME"));

                        //3.category_3
                        JSONObject category_3 = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_CATEGORY3, ps.getCategory3_id());
                        ps.setCategory3_name(category_3.getString("NAME"));

                        //4.trademark
                        JSONObject tm = DimUtil.readDim(phoenixCon, jedis, Constants.DIM_BASE_TRADEMARK, ps.getTm_id());
                        ps.setTm_name(tm.getString("TM_NAME"));

                        //将宽表输出到流
                        resultFuture.complete(Collections.singletonList(ps));
                    }
                },
                30L,
                TimeUnit.SECONDS
        );
    }

    private SingleOutputStreamOperator<ProductStats> aggregateStream(DataStream<ProductStats> unionStream) {
        return unionStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy

                                .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((ele, ts) -> ele.getTs())
                )
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(
                        new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats ps1,
                                                       ProductStats ps2) {
                                //基本聚集指标
                                ps1.setDisplay_ct(ps1.getDisplay_ct() + ps2.getDisplay_ct());
                                ps1.setClick_ct(ps1.getClick_ct() + ps2.getClick_ct());
                                ps1.setFavor_ct(ps1.getFavor_ct() + ps2.getFavor_ct());
                                ps1.setCart_ct(ps1.getCart_ct() + ps2.getCart_ct());

                                //订单参数
                                ps1.setOrder_amount(ps1.getOrder_amount().add(ps2.getOrder_amount()));
                                ps1.getOrderIdSet().addAll(ps2.getOrderIdSet());
                                ps1.setOrder_ct((long) ps1.getOrderIdSet().size());
                                ps1.setOrder_sku_num(ps1.getOrder_sku_num() + ps2.getOrder_sku_num());
                                //支付参数
                                ps1.setPayment_amount(ps1.getPayment_amount().add(ps2.getPayment_amount()));
                                ps1.getPaidOrderIdSet().addAll(ps2.getPaidOrderIdSet());
                                ps1.setPaid_order_ct((long) ps1.getPaidOrderIdSet().size());

                                //退款参数
                                ps1.setRefund_amount(ps1.getRefund_amount().add(ps2.getRefund_amount()));
                                ps1.getRefundOrderIdSet().addAll(ps2.getRefundOrderIdSet());
                                ps1.setRefund_order_ct((long) ps1.getRefundOrderIdSet().size());

                                //评论
                                ps1.setComment_ct(ps1.getComment_ct() + ps2.getComment_ct());
                                ps1.setGood_comment_ct(ps1.getGood_comment_ct() + ps2.getGood_comment_ct());
                                return ps1;
                            }
                        }
                        , new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<ProductStats> it,
                                                Collector<ProductStats> out) {
                                //更新时间
                                ProductStats ps = it.iterator().next();
                                ps.setStt(MyUtil.toDateTime(ctx.window().getStart()));
                                ps.setEdt(MyUtil.toDateTime(ctx.window().getEnd()));
                                ps.setTs(ctx.currentProcessingTime());
                                out.collect(ps);
                            }
                        });
    }

    private DataStream<ProductStats> parseStreamAndUnion(Map<String, DataStreamSource<String>> srcStreams) {
        //1.点击
        SingleOutputStreamOperator<ProductStats> clickStream = srcStreams
                .get(Constants.TOPIC_DWD_PAGE_LOG)
                .flatMap(new FlatMapFunction<String, ProductStats>() {
                    @Override
                    public void flatMap(String line,
                                        Collector<ProductStats> out) {
                        JSONObject obj = JSON.parseObject(line);
                        JSONObject page = obj.getJSONObject("page");
                        //取字段
                        String action = page.getString("page_id");
                        //page_log中有多种行为:浏览good_detail,加购cart,收藏favor
                        //必须筛选出对商品的"点击事件", 浏览商品good_detail = 点击商品
                        if ("good_detail".equals(action)) {
                            Long sku_id = page.getLong("item");
                            Long ts = obj.getLong("ts");
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setClick_ct(1L);
                            out.collect(ps);
                        }
                    }
                });

        //2.曝光
        SingleOutputStreamOperator<ProductStats> displayStream = srcStreams
                .get(Constants.TOPIC_DWD_DISPLAY_LOG)
                .map(line -> {
                            JSONObject displayInfo = JSON.parseObject(line);
                            Long sku_id = displayInfo.getLong("item");
                            Long ts = displayInfo.getLong("ts");
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setDisplay_ct(1L);
                            return ps;
                        }
                );

        //3.收藏
        SingleOutputStreamOperator<ProductStats> favorStream = srcStreams
                .get(Constants.TOPIC_DWD_FAVOR_INFO)
                .map(line -> {
                            JSONObject favorInfo = JSON.parseObject(line);
                            Long sku_id = favorInfo.getLong("sku_id");
                            Long ts = MyUtil.toTs(favorInfo.getString("create_time"));
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setFavor_ct(1L);
                            return ps;
                        }
                );

        //4.加购
        SingleOutputStreamOperator<ProductStats> cartStream = srcStreams
                .get(Constants.TOPIC_DWD_CART_INFO)
                .map(line -> {
                            JSONObject cartInfo = JSON.parseObject(line);
                            Long sku_id = cartInfo.getLong("sku_id");
                            Long ts = MyUtil.toTs(cartInfo.getString("create_time"));
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setCart_ct(1L);
                            return ps;
                        }
                );

        //5.支付
        SingleOutputStreamOperator<ProductStats> paymentStream = srcStreams
                .get(Constants.TOPIC_DWM_PAYMENT_WIDE)
                .map(line -> {
                            PaymentWide paymentWide = JSON.parseObject(line, PaymentWide.class);
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(paymentWide.getSku_id());
                            ps.setTs(MyUtil.toTs(paymentWide.getPayment_create_time()));
                            ps.setPayment_amount(paymentWide.getSplit_total_amount());
                            ps.getPaidOrderIdSet().add(paymentWide.getPayment_id());
                            return ps;
                        }
                );

        //6.订单
        SingleOutputStreamOperator<ProductStats> orderStream = srcStreams
                .get(Constants.TOPIC_DWM_ORDER_WIDE)
                .map(line -> {
                            OrderWide orderWide = JSON.parseObject(line, OrderWide.class);
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(orderWide.getSku_id());
                            ps.setSku_name(orderWide.getSku_name());
                            ps.setOrder_sku_num(orderWide.getSku_num());
                            ps.setTs(MyUtil.toTs(orderWide.getCreate_time()));
                            ps.setOrder_amount(orderWide.getSplit_total_amount());
                            ps.getOrderIdSet().add(orderWide.getOrder_id());
                            return ps;
                        }
                );

        //7.退款
        SingleOutputStreamOperator<ProductStats> refundStream = srcStreams
                .get(Constants.TOPIC_DWD_REFUND_PAYMENT)
                .map(line -> {
                            JSONObject refundInfo = JSON.parseObject(line);
                            Long sku_id = refundInfo.getLong("sku_id");
                            Long ts = MyUtil.toTs(refundInfo.getString("create_time"));
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setRefund_amount(refundInfo.getBigDecimal("total_amount"));
                            ps.getRefundOrderIdSet().add(refundInfo.getLong("order_id"));
                            return ps;
                        }
                );

        //8.评价
        SingleOutputStreamOperator<ProductStats> commentStream = srcStreams
                .get(Constants.TOPIC_DWD_COMMENT_INFO)
                .map(line -> {
                            JSONObject comment = JSON.parseObject(line);
                            Long sku_id = comment.getLong("sku_id");
                            Long ts = MyUtil.toTs(comment.getString("create_time"));
                            String appraise = comment.getString("appraise");
                            //封装为ProductStats
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(sku_id);
                            ps.setTs(ts);
                            ps.setComment_ct(1L);
                            //好评
                            Long good_ct = Constants.APPRAISE_GOOD.equals(appraise) ? 1L : 0L;
                            ps.setGood_comment_ct(good_ct);
                            return ps;
                        }
                );
        return clickStream.union(
                displayStream,
                favorStream,
                cartStream,
                paymentStream,
                orderStream,
                refundStream,
                commentStream);
    }
}
